package com.zhao.demo.bound.demo04_srctarget_hdfs

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.core.fs.FileSystem

/**
 * Description: 计算指定目录下单词出现的次数(源目录和目的地目录都在hdfs)<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/11/25 9:15
 *
 * @author 柒柒
 * @version : 1.0
 */

object BoundedFlowDemo3 {
  def main(args: Array[String]): Unit = {

    //1.拦截非法的参数
    if (args == null || args.length != 4){
      println(
        """
          |警告!输入的参数不符合要求! --input <源文件path> --output <目的地path>
          |""".stripMargin)
      System.exit(-1)
    }

    //2.从参数里面获得源文件path和目的地path
    val tool: ParameterTool = ParameterTool.fromArgs(args)
    val inputPath = tool.get("input")
    val outputPath = tool.get("output")

    //3.环境配置
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    //4.计算,打印或者保存结果
    //a)导入单例类scala中隐式成员
    import org.apache.flink.api.scala._

    //b)保存
    env.readTextFile(inputPath)
      .flatMap(_.split("\\s+"))
      .filter(_.nonEmpty)
      .map((_,1))
      .groupBy(0)
      .sum(1)
      .writeAsText(outputPath,FileSystem.WriteMode.OVERWRITE)
      .setParallelism(1)

    //c)正式的去执行
    env.execute(this.getClass.getSimpleName)


  }
}
